package com.navinfo.opentsp.platform.computing.analysis.application;

import com.navinfo.opentsp.platform.computing.analysis.entity.disconnectedVehicle.OutStatusDto;
import com.navinfo.opentsp.platform.computing.analysis.entity.disconnectedVehicle.PointForDisconnectAnalysis;
import com.navinfo.opentsp.platform.computing.analysis.entity.disconnectedVehicle.ProcessedInfo;
import com.navinfo.opentsp.platform.computing.analysis.service.DisconnectedVehicleAlgorithmService;
import com.navinfo.opentsp.platform.computing.analysis.service.DisconnectedVehicleLoadDataService;
import com.navinfo.opentsp.platform.computing.analysis.service.DisconnectedVehicleSaveService;
import com.navinfo.opentsp.platform.computing.analysis.service.DisconnectedVehicleSortGroupService;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 异常断连/长停断连，断连原因判断
 */
public class DisconnectedVehicleApplication {
    private static Logger logger = LoggerFactory.getLogger(DisconnectedVehicleApplication.class);

    public static void main(String[] args) throws Exception {
        final String day = args[0];
        SparkSession spark = getSparkSession();
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        _printLog("spark initiation complete");

        DisconnectedVehicleLoadDataService loadDataService = getLoadDataService();
        DisconnectedVehicleSortGroupService groupService = getGroupSortService();
        DisconnectedVehicleAlgorithmService algorithmService = new DisconnectedVehicleAlgorithmService(day);
        DisconnectedVehicleSaveService saveService = getSaveService();
        _printLog("service initiation complete");

        // 读取数据
        _printLog("loading data [1/3]");
        /* 当日断连的终端号集合，记为{T1} */
        JavaRDD<Long> terminalIds = loadDataService.loadTerminalIds(spark, day);
        _printLog("loading data [2/3]");
        /* 断连信息表中终端号集合，记为{T2} */
        JavaRDD<ProcessedInfo> processedInfoTids = loadDataService.loadDisconnectionInfo(spark);
        /* 断连信息表中状态='连接'的终端号集合，记为{T2a} */
        JavaRDD<Long> tid2a = processedInfoTids.filter(p -> p.getConnected() == 1).map(ProcessedInfo::getTid);
        /* 断连信息表中状态='断连'的终端号集合，记为{T2b} */
        JavaRDD<Long> tid2b = processedInfoTids.filter(p -> p.getConnected() == 0).map(ProcessedInfo::getTid);
        /*
         * 更新断连原因的终端号：(在T1中存在 && 在T2a中存在) 或 (在T1中存在 && 在T2中不存在)
         *               等价于：在T1中存在 && 在T2b中不存在 <=> T1.subtract(T2b) <=> 图中数字1包围部分
         * 需要更新状态的终端号：在T1中不存在 && 在T2b中存在 <=> T2b.subtract(T1) <=> 图中数字2包围部分
         * 需要保持原样的终端号：(在T1中存在 && 在T2b中存在) 或 (在T1中不存在 && 在T2a中存在)
         *               <=> T1.intersection(T2b).union(T2a.subtract(T1)) <=> 图中点状部分
         * +-----------------------+
         * | <T2a> . . . . . . . . |
         * | . . . . . . . . . . . |
         * | . . . . . .+1111111111+1111111111111+
         * | . . . . . .1          |        <T1> 1
         * | . . . . . .1          |             1
         * | . . . . . .1          |             1
         * +222222222222+1111111111+             1
         * 2 <T2b>      2 . . . . .1             1
         * 2            2 . . . . .1             1
         * 2            2 . . . . .1             1
         * 2            +2222222222+1111111111111+
         * 2                       2
         * 2                       2
         * +22222222222222222222222+
         */
        _printLog("loading data [3/3]");
        JavaRDD<PointForDisconnectAnalysis> points = loadDataService.loadPoint(spark, jsc, terminalIds.subtract(tid2b), day);

        // 分组排序
        _printLog("sorting data");
        JavaPairRDD<Long, List<PointForDisconnectAnalysis>> pointGroup = groupService.sortAndGroup(points);

        // 计算更新断连原因的终端号
        _printLog("calculating [1/3]");
        JavaRDD<ProcessedInfo> saveRdd1 = pointGroup.mapPartitions((FlatMapFunction<Iterator<Tuple2<Long, List<PointForDisconnectAnalysis>>>, ProcessedInfo>) tuple2Iterator ->
                new Iterator<ProcessedInfo>() {
                    @Override
                    public boolean hasNext() {
                        return tuple2Iterator.hasNext();
                    }

                    @Override
                    public ProcessedInfo next() {
                        Tuple2<Long, List<PointForDisconnectAnalysis>> tuple2 = tuple2Iterator.next();
                        Long tid = tuple2._1;
                        List<PointForDisconnectAnalysis> points = tuple2._2;

                        OutStatusDto outStatus = new OutStatusDto();
                        if (points != null && points.size() > 0) {
                            int lastIndex = points.size() - 1;
                            for (int i = 0; i < points.size(); i++) {
                                outStatus.setLast(i == lastIndex);
                                algorithmService.dealPoint(points.get(i), outStatus);
                            }
                        } else {
                            logger.error("原始点数据为空");
                        }

                        ProcessedInfo result = new ProcessedInfo();
                        result.setTid(tid);
                        result.setConnected(0);
                        result.setLackOfElectricity(outStatus.isLackOfElectricity() ? 1 : 0);
                        result.setModified(outStatus.isModified() ? 1 : 0);
                        result.setPoorSignal(outStatus.isPoorSignal() ? 1 : 0);
                        result.setAbnormalDisconnection(outStatus.isAbnormalDisconnection() ? 1 : 0);
                        result.setLongStayDisconnection(outStatus.isLongStayDisconnection() ? 1 : 0);
                        result.setAlarm(outStatus.getAlarm());
                        return result;
                    }
                });

        // 计算需要更新状态的终端号
        _printLog("calculating [2/3]");
        final Broadcast<Map<Long, ProcessedInfo>> processedInfoBc = broadcast(jsc, processedInfoTids);
        JavaRDD<ProcessedInfo> saveRdd2 = tid2b.subtract(terminalIds).mapPartitions((FlatMapFunction<Iterator<Long>, ProcessedInfo>) longIterator ->
                new Iterator<ProcessedInfo>() {
                    @Override
                    public boolean hasNext() {
                        return longIterator.hasNext();
                    }

                    @Override
                    public ProcessedInfo next() {
                        Long tid = longIterator.next();
                        ProcessedInfo processedInfo = processedInfoBc.getValue().get(tid);
                        return copyOf(processedInfo, 1);
                    }
                });

        // 计算需要更保持原样的终端号
        _printLog("calculating [3/3]");
        JavaRDD<ProcessedInfo> saveRdd3 = terminalIds.intersection(tid2b).union(tid2a.subtract(terminalIds)).mapPartitions((FlatMapFunction<Iterator<Long>, ProcessedInfo>) longIterator ->
                new Iterator<ProcessedInfo>() {
                    @Override
                    public boolean hasNext() {
                        return longIterator.hasNext();
                    }

                    @Override
                    public ProcessedInfo next() {
                        Long tid = longIterator.next();
                        ProcessedInfo processedInfo = processedInfoBc.getValue().get(tid);
                        return copyOf(processedInfo, null);
                    }
                });

        JavaRDD<ProcessedInfo> saveRdd = saveRdd1.union(saveRdd2).union(saveRdd3);
        saveRdd.persist(StorageLevel.MEMORY_ONLY_SER());
        // 保存落盘
        _printLog("saving data [" + saveRdd.count() + "]");
        saveService.save(spark, saveRdd);

        saveRdd.unpersist();
        jsc.close();
        spark.stop();
    }

    private static void _printLog(String s) {
        logger.error(s);
    }

    private static SparkSession getSparkSession() {
        SparkSession spark = null;
        String appName = "DisconnectedVehicleAnalysis";
        String master = PropertiesUtil.getProperties("spark.master");
        if ("local".equals(master)) {
            spark = SparkSession.builder()
                    .appName(appName)
                    .master(master)
                    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                    .config("spark.sql.warehouse.dir", "hdfs://platform-010-030-050-032:8020/apps/hive/warehouse")
//                    .config("spark.sql.hive.metastore.jars", "maven")
                    .enableHiveSupport()
                    .getOrCreate();
        } else if ("yarn".equals(master)) {
            spark = SparkSession.builder()
                    .appName(appName)
                    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                    .enableHiveSupport()
                    .getOrCreate();
        }
        return spark;
    }

    private static DisconnectedVehicleLoadDataService getLoadDataService() {
        String pointDbName = PropertiesUtil.getProperties("disconnectedVehicle.hive.point.dbname");
        String pointTableName = PropertiesUtil.getProperties("disconnectedVehicle.hive.point.tbname");
        String saveDbName = PropertiesUtil.getProperties("hive.save.disconnectedVehicle.db");
        return new DisconnectedVehicleLoadDataService(pointDbName, pointTableName, saveDbName);
    }

    private static DisconnectedVehicleSortGroupService getGroupSortService() {
        return new DisconnectedVehicleSortGroupService(null);
    }

    private static DisconnectedVehicleSaveService getSaveService() {
        String dbName = PropertiesUtil.getProperties("hive.save.disconnectedVehicle.db");
        String tableName = PropertiesUtil.getProperties("hive.save.disconnectedVehicle.table");
        return new DisconnectedVehicleSaveService(dbName, tableName);
    }

    private static Broadcast<Map<Long, ProcessedInfo>> broadcast(JavaSparkContext jsc, JavaRDD<ProcessedInfo> processedInfo) {
        Map<Long, ProcessedInfo> data = new HashMap<>();
        processedInfo.collect().forEach(p -> data.put(p.getTid(), p));
        return jsc.broadcast(data);
    }

    private static ProcessedInfo copyOf(ProcessedInfo processedInfo, Integer connected) {
        ProcessedInfo result = new ProcessedInfo();
        result.setTid(processedInfo.getTid());
        if (connected != null) {
            result.setConnected(connected);
        } else {
            result.setConnected(processedInfo.getConnected());
        }
        result.setLackOfElectricity(processedInfo.getLackOfElectricity());
        result.setModified(processedInfo.getModified());
        result.setPoorSignal(processedInfo.getPoorSignal());
        result.setAbnormalDisconnection(processedInfo.getAbnormalDisconnection());
        result.setLongStayDisconnection(processedInfo.getLongStayDisconnection());
        result.setAlarm(processedInfo.getAlarm());
        return result;
    }
}
